package mapreduce;

import java.io.File;
import java.io.IOException;
import java.util.Vector;


public class MapTaskConf<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends TaskConf {

	private String FileAddress;
	private Class<?> MapClass;
	private int numofreduce;
	private Class<?> MapInputKeyClass;
	private Class<?> MapInputValueClass;
	private Manager<KEYOUT, VALUEOUT> manager;
	public MapTaskConf(Class<?> MapClass, Class<?> inputkeyclass,
			Class<?> inputvalueclass, String Jobdir, int Taskid, int numofreduce, boolean mflag, Vector<pair<KEYOUT, VALUEOUT>>[][] middleBuffer)
			throws IOException {
		super(Taskid,mflag);
		this.MapClass = MapClass;
		this.FileAddress = Jobdir + "map/input/" + Taskid + ".data";
		this.numofreduce = numofreduce;
		this.MapInputKeyClass = inputkeyclass;
		this.MapInputValueClass = inputvalueclass;
		File mapresultdir = new File(Jobdir + "map/result/");
		if (!mapresultdir.exists())
			mapresultdir.mkdirs();
		if(mflag==false)
			manager = new MapDiskManager<KEYOUT, VALUEOUT>(Jobdir, Taskid,
				numofreduce);
		else {
//			System.out.println("taskconf map len="+middleBuffer.length);
			manager = new MapBufferManager<KEYIN, VALUEIN, KEYOUT, VALUEOUT>(Taskid, middleBuffer, numofreduce);
		}
	}

	public void write(KEYOUT key, VALUEOUT value) throws IOException {
		manager.write(key, value);

	}

	public void close() throws IOException {
		manager.close();
	}

	public int getnumofreduce() {
		return this.numofreduce;
	}

	public String getFileAddress() {
		return FileAddress;
	}

	public Class<?> getInputKeyClass() {
		return this.MapInputKeyClass;
	}

	public Class<?> getInputValueClass() {
		return this.MapInputValueClass;
	}

	public void setFileAddress(String FileAddress) {
		this.FileAddress = FileAddress;
	}

	public Class<?> getMapClass() {
		return MapClass;
	}

	public void setMapClass(Class<?> MapClass) {
		this.MapClass = MapClass;
	}

}
